<?php
require_once 'Stomp.php';
require_once 'Stomp/Message.php';
require_once 'Stomp/Message/Map.php';

require_once 'TVB/Config/ConfigReader.php';
require_once 'TVB/Config/ConfigReaderFactory.php';

abstract class BaseConsumer {
	protected $config;
	protected $selector;
	
	public function __construct($configFile, $name, $selector = null, $broker = null, $queue = null) {
		if ($broker != null) {
			$this->config->broker = $broker;
			$this->config->queue = $queue;
		} else {
			$config= ConfigReaderFactory::getInstance()->getConfigReader($configFile)->getConfig();
			$consumerConfig = $config->xpath("/config/consumer[@name='{$name}']");
			$this->config = $consumerConfig[0];
		}
		$this->selector = $selector;
	}
	
	protected function sendErrorQueue(Stomp_Frame $msg) {
		$broker = (string) $this->config->broker;
		$queue = (string) $this->config->queue;
		$queue .= 'Error';
		$con = new Stomp($broker);
		$con->connect();
		$errorMsg = new Stomp_Message_Map($msg->map);
		$ret = $con->send($queue, $errorMsg, array('persistent' => 'true'), true);
		$con->disconnect();
		
		return $ret;
	}
	
	public function run() {
		$broker = (string) $this->config->broker;
		$queue = (string) $this->config->queue;
		$ret = TRUE;
		while ($ret) {
			$con = new Stomp($broker);
			$con->connect();
			if (is_null($this->selector)) {
				$con->subscribe($queue);
			} else {
				$con->subscribe($queue, array('selector' => $this->selector));
			}
		
			$msg = $con->readFrame();
			if ($msg === FALSE) {
				error_log(date('Y-m-d H:i:s') . ": no message received");

				if ( class_exists('MySQLConnectionPool') ) {
					MySQLConnectionPool::clearPool();
				}

				if ( class_exists('DBConnPoolFactory') ) {
					DBConnPoolFactory::getInstance()->clearConnPool();
					DBConnPoolFactory::getInstance()->clearFarmConnPool();
				}
				continue;
			}

			
			if ($msg->command != 'MESSAGE') {
				continue;
			}
			
			if (@$msg->headers['compression'] == "bzip") {
                $base64 = $msg->body;
                $bzip = base64_decode($base64);
                $json = bzdecompress($bzip);
                $msg->map = json_decode($json);
            }

			
			$ret = $this->process($msg);
			if ($ret) {
				$ret = $con->ack($msg);
			}

			$con->disconnect();
		}
	}

	abstract protected function process(Stomp_Frame $msg);
	
}

?>
